弹力设计篇之"重试设计" - [2026重制版]

<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-retry</artifactId>
    <version>2.1.0</version>
</dependency>
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-spring-boot2</artifactId>
    <version>2.1.0</version>
</dependency>

resilience4j:
  retry:
    instances:
      # 基础重试配置（用于一般远程调用）
      backendService:
        maxAttempts: 3                    # 最大重试次数（含首次调用）
        waitDuration: 500ms               # 初始等待时间
        enableExponentialBackoff: true     # 启用指数退避
        exponentialBackoffMultiplier: 2.0   # 退避乘数
        randomizationFactor: 0.5           # 抖动因子（0-0.5）
        retryExceptions:                   # 触发重试的异常
          - java.net.SocketTimeoutException
          - java.net.ConnectException
          - org.springframework.web.client.HttpServerErrorException
          - io.github.resilience4j.circuitbreaker.CallNotPermittedException
        ignoreExceptions:                  # 忽略不重试的异常
          - org.springframework.web.client.HttpClientErrorException  # 4xx错误
          - java.lang.IllegalArgumentException
          - com.example.business.BusinessException
        
      # 支付服务的特殊配置（更保守的重试策略）
      paymentGateway:
        maxAttempts: 2                    # 支付只允许重试1次（共2次尝试）
        waitDuration: 1000ms              # 初始等待1秒
        enableExponentialBackoff: true
        exponentialBackoffMultiplier: 3.0   # 更激进的退避（3x）
        randomizationFactor: 0.6
        retryOnResultPredicate:            # 基于返回结果判断是否重试
          io.github.resilience4j.retry.RetryConfig$$Lambda$1
        retryExceptions:
          - com.example.payment.PaymentTimeoutException
          - com.example.payment.GatewayBusyException
          
      # 数据库查询（快速失败策略）
      databaseQuery:
        maxAttempts: 1                    # 不重试，快速失败
        waitDuration: 0ms

@Service
@RequiredArgsConstructor
@Slf4j
public class BackendServiceClient {
 
    private final BackendServiceFeignClient feignClient;
    private final RetryRegistry retryRegistry;
 
    /**
     * 方式一：注解方式（推荐用于简单场景）
     */
    @Retry(name = "backendService", fallbackMethod = "fallbackGetData")
    public DataResponse getData(String id) {
        log.info("Fetching data for id: {}", id);
        return feignClient.getData(id);
    }
 
    /**
     * 降级方法（所有重试耗尽后的兜底）
     */
    public DataResponse fallbackGetData(String id, Exception ex) {
        log.warn("All retries exhausted for id={}, error={}", id, ex.getMessage());
        
        # 尝试从缓存获取或返回默认值
        return DataResponse.fromCacheOrDefault(id);
    }
 
    /**
     * 方式二：编程式方式（更灵活的控制）
     */
    public PaymentResult processPayment(PaymentRequest request) {
        # 获取配置好的Retry实例
        Retry retry = retryRegistry.retry("paymentGateway");
 
        # 使用Supplier装饰器包装调用
        Supplier<PaymentResult> supplier = Retry.decorateSupplier(retry, () -> {
            log.info("Attempting payment processing...");
            PaymentResult result = feignClient.charge(request);
            
            # 自定义检查：基于业务逻辑决定是否需要重试
            if (result.isPending()) {
                throw new PaymentStillProcessingException("Payment is still processing");
            }
            
            return result;
        });
 
        try:
            return supplier.get();
        } catch (MaxRetriesExceededException e):
            log.error("Payment retry limit exceeded", e);
            return PaymentResult.failed("Service temporarily unavailable");
        }
}

@FeignClient(
    name = "inventory-service",
    configuration = InventoryFeignConfig.class,
    fallbackFactory = InventoryFallbackFactory.class
)
public interface InventoryFeignClient {
 
    @GetMapping("/api/inventory/{productId}")
    @CircuitBreaker(name = "inventory")  # 熔断器
    @RateLimiter(name = "inventory")     # 限流器
    @Retry(name = "inventoryBackend")     # 重试器
    InventoryResponse checkStock(@PathVariable String productId);
 
    @PutMapping("/api/inventory/reserve")
    @Retry(name = "inventoryBackend")
    ReserveResult reserveStock(@RequestBody ReserveRequest request);
}
 
/**
 * Feign配置类
 */
@Configuration
public class InventoryFeignConfig {
 
    @Bean
    public RequestInterceptor requestInterceptor() {
        return template -> {
            # 注入Trace ID用于链路追踪
            String traceId = MDC.get("traceId");
            if (traceId != null) {
                template.header("X-Trace-Id", traceId);
            }
            
            # 注入幂等性Key
            String idempotencyKey = IdempotencyContext.getKey();
            if (idempotencyKey != null) {
                template.header("X-Idempotency-Key", idempotencyKey);
            }
        };
    }
 
    @Bean
    public ErrorDecoder errorDecoder() {
        return new CustomErrorDecoder();
    }
}
 
/**
 * 自定义ErrorDecoder：区分可重试和不可重试的错误
 */
public class CustomErrorDecoder implements ErrorDecoder {
 
    private final Logger log = LoggerFactory.getLogger(CustomErrorDecoder.class);
 
    @Override
    public Exception decode(String methodKey, Response response) {
        int status = response.status();
 
        if (status >= 400 && status < 500):
            # 4xx错误：不应重试（参数问题、权限问题等）
            if (status == 429):  # Too Many Requests 可重试
                return new TooManyRequestsException(status, "Rate limited");
            return new ClientException(status, "Client error, no retry");
 
        elif (status >= 500):
            # 5xx错误：可重试
            switch (status):
                case 503:  # Service Unavailable
                    return new ServiceUnavailableException(status, "Service unavailable");
                case 504:  # Gateway Timeout
                    return new GatewayTimeoutException(status, "Gateway timeout");
                default:
                    return new ServerException(status, "Server error, should retry");
 
        return new RuntimeException("Unknown HTTP status: " + status);
    }
}

/**
 * 自适应重试策略实现
 * 
 * 思路参考TCP的拥塞控制算法：
 * - 成功时：线性增加（Additive Increase）
 * - 失败时：乘法减少（Multiplicative Decrease）
 * - 加入随机抖动避免同步
 */
public class AdaptiveRetryStrategy implements RetryStrategy {
 
    private static final long MIN_WAIT_MS = 100;       // 最小等待时间
    private static final long MAX_WAIT_MS = 30000;      // 最大等待时间（30秒）
    private static final double INCREASE_FACTOR = 1.5;  # 成功时增加因子
    private static final double DECREASE_FACTOR = 0.5;  # 失败时减少因子
    private static final double JITTER_FACTOR = 0.2;    // 抖动因子
 
    private final Map<String, RetryState> stateMap = new ConcurrentHashMap<>();
 
    @Override
    public long calculateWaitTime(String targetService, int attemptNumber, boolean lastSuccess) {
        RetryState state = stateMap.computeIfAbsent(targetService, k -> new RetryState());
 
        if (lastSuccess) {
            # 成功：增加等待时间（但不超过最大值）
            state.currentWaitMs = Math.min(
                (long)(state.currentWaitMs * INCREASE_FACTOR),
                MAX_WAIT_MS
            );
            state.consecutiveFailures = 0;
        } else {
            # 失败：大幅减少等待时间
            state.currentWaitMs = Math.max(
                (long)(state.currentWaitMs * DECREASE_FACTOR),
                MIN_WAIT_MS
            );
            state.consecutiveFailures++;
        }
 
        # 加入随机抖动（避免多个客户端同时重试造成"惊群效应"）
        long jitter = (long)(state.currentWaitMs * JITTER_FACTOR * (Math.random() * 2 - 1));
        long finalWait = Math.max(MIN_WAIT_MS, state.currentWaitMs + jitter);
 
        log.debug("[AdaptiveRetry] service={}, attempt={}, wait={}ms, consecutiveFailures={}",
                 targetService, attemptNumber, finalWait, state.consecutiveFailures);
 
        return finalWait;
    }
 
    @Data
    private static class RetryState {
        private long currentWaitMs = 1000;  # 初始等待1秒
        private int consecutiveFailures = 0;
    }
}

// proto文件中的service定义
service OrderService {
  // 定义重试策略（在proto层面）
  rpc CreateOrder(CreateOrderRequest) returns (CreateOrderResponse) {
    option (google.api.http) = {
      post: "/v1/orders"
      body: "*"
    };
    // gRPC重试策略注解
    option (grpc.retry_policy) = {
      max_attempts: 3
      initial_backoff: "0.1s"
      max_backoff: "1s"
      backoff_multiplier: 2.0
      retryable_status_codes: [UNAVAILABLE, DEADLINE_EXCEEDED, RESOURCE_EXHAUSTED]
    };
  }
}

ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9090)
    .usePlaintext()
    .enableRetry()  // 启用重试
    .maxRetryAttempts(3)
    .retryWithBackoff(
        InitialBackoff.seconds(1),
        MaxBackoff.seconds(10),
        BackoffMultiplier.valueOf(2),
        Jitter.valueOf(0.2)
    )
    .retryOnStatus(Status.Code.UNAVAILABLE)
    .retryOnStatus(Status.Code.DEADLINE_EXCEEDED)
    .build();